package report

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}
import utils.CleanDataUtils

/**
  * Created by 王康 on 2018/6/30.
  */
object Media_analysis_Core {
  def main(args: Array[String]) {
    val conf: SparkConf = new SparkConf().setAppName("Media_analysis").setMaster("local[4]")
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    val ssc: SparkContext = new SparkContext(conf)
    val context: SQLContext = new SQLContext(ssc)

    //整理出AppName集合
    val file: RDD[String] = ssc.textFile("E:\\大数据资料\\project2\\app_dict.txt")
    val collect: Array[(String, String)] = file.map(t => t.split("\t")).filter(t =>
      t.length >= 5
        && !t(4).isEmpty)
      .map(t => (t(4), t(1))).collect()
    val appName: Map[String, String] = collect.toMap
    /* println(appName.toBuffer)*/
    //制造一个广播变量
    val bdt: Broadcast[Map[String, String]] = ssc.broadcast(appName)


    val parquet: DataFrame = context.read.parquet("E:\\大数据资料\\project2\\parquet_1.6.3test")
    val rdd: RDD[Row] = parquet.rdd
    val filterdata: RDD[(String, (Int, Int, Int, Int, Int, Int, Int, Double, Double))] = rdd.map(row => {
      val appNameMap: Map[String, String] = bdt.value
      val appid: String = row.getString(13)
      var appname: String = row.getString(14)
      if (appname.isEmpty) {
        //通过appid 去map中 找 如果能找到值 就返回res 找不到就返回unkonw 同时 如果找到的res 为空字符串 就用appid代替
        val res: String = appNameMap.getOrElse(appid, "unkonw")
        if (res.isEmpty) {
          println("找不到App名字 用appid代替")
          appname = appid
        } else {
          appname = res
          println("找到APP名字 用APPdict代替")
        }
      }

      val (winprice: Double, adpayment: Double, flag1: Boolean,
      flag2: Boolean, flag3: Boolean, flag4: Boolean, flag5: Boolean,
      flag6: Boolean, flag7: Boolean)=CleanDataUtils.cleanDataByAppNameCore(row)

      (appname, (if (flag1) 1 else 0, if (flag2) 1 else 0, if (flag3) 1 else 0, if (flag4) 1 else 0, if (flag5) 1 else 0, if (flag6) 1 else 0, if (flag7) 1 else 0, if (flag5) winprice else 0.0, if (flag5) adpayment else 0.0))
    })
    val result: RDD[(String, (Int, Int, Int, Int, Int, Int, Int, Double, Double))] = filterdata.reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2, x._3 + y._3, x._4 + y._4, x._5 + y._5, x._6 + y._6, x._7 + y._7, x._8 + y._8, x._9 + y._9))
    /*  result.saveAsTextFile("E:\\大数据资料\\project2\\Media_analysis")*/
    result.foreach(t => println(t))
  }


}

